package com.bcqn.task;

import com.alibaba.fastjson.JSONObject;
import com.bcqn.service.PostReviewService;
import com.bcqn.service.PostService;
import io.lettuce.core.KeyScanCursor;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.ScanArgs;
import io.lettuce.core.ScanCursor;
import io.lettuce.core.api.async.RedisAsyncCommands;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * LikeTask 点赞数据定时刷库
 *
 * @author Kelly
 * @version v1.0
 * @date 2020/4/22 22:18
 */
@Configuration
@EnableScheduling
public class LikeTask {
    private static Logger logger = LoggerFactory.getLogger(LikeTask.class);

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private PostService postService;

    @Autowired
    private PostReviewService postReviewService;
    /**
     * 帖子点赞定时刷库
     */
//    @Scheduled(cron = "0/10 * * * * *")
    public void postLikeFlushDB() {
        long start = System.currentTimeMillis();
        String pattern = "*"+"POSTID-"+"*";
        // 获取所有的keys
        List<String> keys = scanKeys(redisTemplate,pattern);
        long end = System.currentTimeMillis();
        logger.info("获取redis中所有key值所用时间：{}ms",end-start);
        long s = System.currentTimeMillis();
        // 根据keys获取values
        List<Object> list = redisTemplate.executePipelined((RedisCallback<String>) connection -> {
            connection.openPipeline();
            for (String key : keys) {
                connection.get(key.getBytes());
            }
            return null;
        });
        long e = System.currentTimeMillis();
        logger.info("获取redis中所有value值所用时间：{}ms",e-s);
        Map<Long,Integer> map = buildListMap(keys,list);
        if (!map.isEmpty()) {
            // 批量插入数据库
            logger.info(JSONObject.toJSONString(map));
            postService.batchUpdatePostLike(map);
        }
    }
    /**
     * 评论点赞定时刷库
     */
//    @Scheduled(cron = "0/10 * * * * *")
    public void reviewLikeFlushDB() {
        long start = System.currentTimeMillis();
        String pattern = "*"+"REVIEWID-"+"*";
        // 获取所有的keys
        List<String> keys = scanKeys(redisTemplate,pattern);
        long end = System.currentTimeMillis();
        logger.info("获取redis中所有key值所用时间：{}ms",end-start);
        long s = System.currentTimeMillis();
        // 根据keys获取values
        List<Object> list = redisTemplate.executePipelined((RedisCallback<String>) connection -> {
            connection.openPipeline();
            for (String key : keys) {
                connection.get(key.getBytes());
            }
            return null;
        });
        long e = System.currentTimeMillis();
        logger.info("获取redis中所有value值所用时间：{}ms",e-s);

        Map<Long,Integer> map = buildListMap(keys,list);
        if (!map.isEmpty()) {
            // 批量插入数据库
            logger.info(JSONObject.toJSONString(map));
            postReviewService.batchUpdatePostReviewLike(map);
        }
    }
    /**
     * 获取redis中符合pattern表达式的keys
     * @param pattern
     * @return
     */
    private static List<String> scanKeys(RedisTemplate redisTemplate, String pattern) {
        return (List<String>) redisTemplate.execute(connection -> {
            //scan 迭代遍历键，返回的结果可能会有重复，需要客户端去重复
            Set<String> redisKeys = new HashSet<>();
            //lettuce 原生api
            RedisAsyncCommands conn = (RedisAsyncCommands) connection.getNativeConnection();
            //游标
            ScanCursor curs = ScanCursor.INITIAL;
            try {
                //采用 SCAN 命令，迭代遍历所有key
                while (!curs.isFinished()) {
                    long count = 10000L;
                    ScanArgs args = ScanArgs.Builder.matches(pattern).limit(count);
                    RedisFuture<KeyScanCursor<byte[]>> future = conn.scan(curs, args);
                    KeyScanCursor<byte[]> keyCurs = future.get();
                    List<byte[]> ks = keyCurs.getKeys();
                    Set<String> set = ks.stream().map(bytes -> new String(bytes, StandardCharsets.UTF_8)).collect(Collectors.toSet());
                    redisKeys.addAll(set);
                    curs = keyCurs;
                }
            } catch (InterruptedException | ExecutionException e) {
                throw new RuntimeException(e);
            }
            return new ArrayList<>(redisKeys);
        }, true);
    }

    /**
     * 组装点赞key-value的map用于插入数据库
     * @param keys
     * @param list
     * @return
     */
    private Map<Long,Integer> buildListMap(List<String> keys, List<Object> list) {
        Map<Long,Integer> map = new HashMap<>(16);
        for (int i = 0; i < keys.size(); i++) {
            if (list.get(i) != null) {
                String mapKeyStr = StringUtils.substringAfterLast(keys.get(i),"-");
                if (!StringUtils.isBlank(mapKeyStr)) {
                    map.put(Long.parseLong(mapKeyStr),Integer.parseInt(list.get(i).toString()));
                }
            }
        }
        return map;
    }
}
